<?php

/**
 * 生产者
 * */

namespace App\Lib;

use AMQPConnection;
use AMQPChannel;
use AMQPExchange;
use AMQPQueue;
use AMQPConnectionException;
use App\Http\Model\Hint;

class Publisher
{
    public $requestData = [];// 接收参数

    protected $queueConfig = [];// 队列总配置
    protected $queueRoute = [];// 队列路由
    protected $connectConfig = [];// 连接配置

    private static $instance;

    private function __construct()
    {
        $this->queueConfig = config('queue') ?? [];

        /*$this->connectConfig = [
            'host' => env('RABBITMQ_HOST', '127.0.0.1'),
            'port' => env('RABBITMQ_PORT', 5672),
            'vhost' => env('RABBITMQ_VHOST', '/'),
            'login' => env('RABBITMQ_LOGIN', 'guest'),
            'password' => env('RABBITMQ_PASSWORD', 'guest'),
        ];*/

        $this->connectConfig = [
            'host' => $this->queueConfig['RABBITMQ_HOST'] ?? '127.0.0.1',
            'port' => $this->queueConfig['RABBITMQ_PORT'] ?? 5672,
            'vhost' => $this->queueConfig['RABBITMQ_VHOST'] ?? '/',
            'login' => $this->queueConfig['RABBITMQ_LOGIN'] ?? 'guest',
            'password' => $this->queueConfig['RABBITMQ_PASSWORD'] ?? 'guest',
        ];
    }

    /**
     * 返回实例
     * */
    static public function getInstance()
    {
        self::$instance = new self();

        return self::$instance;
    }

    /**
     * 发布
     *
     * param queue_route 队列路由
     * param exchange_type 交换类型 默认topic主题交换  direct:直接交换 topic:主题交换 fanout:
     * param data 数据
     *
     * return array
     * */
    public function publisher()
    {
        $code = Hint::SUCCESS;
        $msg = Hint::getErrCodeMsg($code);
        $data = [];
        do {
            try {
                // 参数判断
                if (empty($this->requestData['queue_route']) or empty($this->requestData['data'])) {
                    $code = Hint::ERR_PARAMS_CHECK_FAIL;
                    $msg = Hint::getErrCodeMsg($code);
                    break;
                }
                // 队列配置项 路由配置 交换类型
                if (empty($this->queueConfig) or empty($this->queueConfig['queueRoute'])) {
                    $code = Hint::ERR_FAIL;
                    $msg = '队列配置项初始' . Hint::getErrCodeMsg($code);
                    break;
                }

                if (!array_key_exists($this->requestData['queue_route'], $this->queueConfig['queueRoute'])) {
                    $code = Hint::ERR_PARAMS_CHECK_FAIL;
                    $msg = '队列路由:' . Hint::getErrCodeMsg($code);
                    break;
                }

                // TODO 队列路由
                $this->queueRoute = $this->queueConfig['queueRoute'][$this->requestData['queue_route']];

                /*if (!extension_loaded('amqp')) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'No extension amqp loaded failed !';
                    break;
                }*/

                $conn = new AMQPConnection($this->connectConfig);
                $conn->connect();
                if (!$conn->isConnected()) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'rabbit-mq Connection failed !';
                    break;
                }

                $channel = new AMQPChannel($conn);
                if (!$channel->isConnected()) {
                    $code = Hint::ERR_FAIL;
                    $msg = 'rabbit-mq Connection through channel failed !';
                    break;
                }

                $exchange = new AMQPExchange($channel);
                $exchange->setFlags(AMQP_DURABLE);// 持久化
                $exchange->setName($this->queueRoute['exchangeName'] ?: '');
                $exchange->setType(AMQP_EX_TYPE_TOPIC); // topic 交换类型
                $exchange->declareExchange();

                //$channel->startTransaction();

                // 设置队列名称
                if (!empty($this->queueRoute['queueName'])) {
                    $queue = new AMQPQueue($channel);
                    $queue->setName($this->queueRoute['queueName']);
                    $queue->setFlags(AMQP_DURABLE);
                    $queue->declareQueue();
                }

                //绑定
                foreach ($this->queueRoute['routeKey'] as $key => $value) {
                    // TODO 队列绑定设置（队列名称不为空，主题键则会发到指定队列上）
                    if (!empty($this->queueRoute['queueName'])) {
                        $queue->bind($this->queueRoute['exchangeName'], $value);
                    }

                    // todo log
                    write_log('publisher', $value . '_' . $key, $this->requestData['data']);

                    //生产消息
                    $ret = $exchange->publish(json_encode($this->requestData['data'], JSON_UNESCAPED_UNICODE), $value, AMQP_MANDATORY, array('delivery_mode' => 2));
                    if (!$ret) {
                        $code = Hint::ERR_FAIL;
                        $msg = '消息发布' . Hint::getErrCodeMsg($code);
                        break;
                    }

                    $data['message'][$value . '_' . $key] = $this->requestData['data'];
                }

                $conn->disconnect();

            } catch (AMQPConnectionException $e) {
                $code = Hint::ERR_SYS_FAIL;
                $msg = Hint::getErrCodeMsg($code) . ':' . $e->getMessage();
                break;
            }

        } while (false);

        return compact('code', 'msg', 'data');
    }
}